package net.datafans.exercise.redis;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;

public class SimpleDaoImpl {
	@Autowired
	private RedisTemplate<Serializable, Serializable> redisTemplate;

	public void setString(final String key, final String val) {
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				connection.set(redisTemplate.getStringSerializer().serialize(key), redisTemplate.getStringSerializer().serialize(val));
				return null;
			}
		});
	}

	public String getString(final String key) {
		return (String) redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				if (connection.exists(redisTemplate.getStringSerializer().serialize(key))) {
					byte[] value = connection.get(redisTemplate.getStringSerializer().serialize(key));
					return redisTemplate.getStringSerializer().deserialize(value);
				}
				return null;
			}
		});
	}

	public void setList(final String key, final String val) {
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				connection.lPush(redisTemplate.getStringSerializer().serialize(key), redisTemplate.getStringSerializer().serialize(val));
				return null;
			}
		});
	}

	@SuppressWarnings("unchecked")
	public List<String> getList(final String key) {
		return (List<String>) redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				if (connection.exists(redisTemplate.getStringSerializer().serialize(key))) {
					List<byte[]> list = connection.lRange(redisTemplate.getStringSerializer().serialize(key), 0, 100);
					List<String> lists = new ArrayList<String>();
					for (byte[] b : list) {
						lists.add(redisTemplate.getStringSerializer().deserialize(b));
					}
					return lists;
				}
				return null;
			}
		});
	}

	public void setMap(final String key, final String field, final String val) {
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				connection.hSet(redisTemplate.getStringSerializer().serialize(key), redisTemplate.getStringSerializer().serialize(field), redisTemplate
						.getStringSerializer().serialize(val));
				return null;
			}
		});
	}

	@SuppressWarnings("unchecked")
	public Map<String, String> getMap(final String key) {
		return (Map<String, String>) redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				if (connection.exists(redisTemplate.getStringSerializer().serialize(key))) {
					Map<byte[], byte[]> map = connection.hGetAll(redisTemplate.getStringSerializer().serialize(key));
					Map<String, String> maps = new HashMap<String, String>();
					Set<Entry<byte[], byte[]>> set = map.entrySet();
					for (Entry<byte[], byte[]> e : set) {
						maps.put(redisTemplate.getStringSerializer().deserialize(e.getKey()), redisTemplate.getStringSerializer().deserialize(e.getValue()));
					}
					return maps;
				}
				return null;
			}
		});
	}
}
